/************************************************************************
Copyright (c) 2010, 2013, C.R. All rights reserved. 

E-mail: crtoyou@gmail.com; crtoyou@163.com

You can redistribute it and/or modify it under the terms of the
GNU Lesser General Public License as published by the
Free Software Foundation version 3 of the License.

This source code is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program. if not, write to the Free Software Foundation, Inc., 
51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
************************************************************************/

#ifndef __CR_IOCP_WITH_THREAD_POOL__
#define __CR_IOCP_WITH_THREAD_POOL__

#include <WinSock2.h>
#include <Windows.h>
#include <vector>
#include <set>
#include <crthread.h>
#include <crdebug.h>

#define IOCP_TIMEOUT  5*60*1000 
#define MONITOR_CYCLE 60*60*1024 //monitor wakes every one hour 

namespace criocp {

using namespace cr;

struct IIocpPortBase
{
	virtual void OnIoCompletion(DWORD dwBytes, LPOVERLAPPED lpOvlp, DWORD dwErr) = 0;
	virtual void OnIocpTimeout(DWORD timeout) = 0;
};


class CIocpCore
{
public:
	CIocpCore(void)
		: m_hIocp(NULL)
		, m_daemon(NULL)
	{
		m_daemon = new CDaemonThread(*this);
	}

	~CIocpCore(void)
	{
		Destroy();
	}

	BOOL Create(int nThreadCount = 0)
	{
		ASSERT(!m_hIocp);
		m_hIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, nThreadCount);
		if (nThreadCount <= 0)
			nThreadCount = AdvisedThreadCount();
		for (int i=0; i<nThreadCount; ++i) {
			CWorkerThread* worker = new CWorkerThread(*this);
			worker->init(true, IOCP_TIMEOUT); //create and start worker
			m_lock.dolock();
			m_vecThread.push_back(worker);
			m_lock.unlock();
		}
		//create and start monitor, do idle every 1 hour of timeout. 
		//paused initialization, to wait the timeout to do idle. 
		if (!m_daemon)
			m_daemon = new CDaemonThread(*this);
		if (m_daemon)
			m_daemon->init(false, MONITOR_CYCLE);
		return TRUE;
	}

	void Destroy()
	{
		//destroy the monitor before destroy worker thread. 
		if (m_daemon) {
			m_daemon->destroy();
			delete m_daemon;
			m_daemon = NULL;
		}
		m_lock.dolock();
		for (int i=0; i<(int)m_vecThread.size(); ++i) {
			CWorkerThread* worker = m_vecThread[i];
			if (worker) {
				worker->destroy();
				delete worker;
			}
		}
		m_vecThread.clear();
		m_lock.unlock();
		if (m_hIocp) {
			CloseHandle(m_hIocp);
			m_hIocp = NULL;
		}
	}

	void AttachPort(IIocpPortBase* pPortObj)
	{
		m_lock.dolock();
		m_regPort.insert(pPortObj);
		m_lock.unlock();
	}

	void DetachPort(IIocpPortBase* pPortObj)
	{
		m_lock.dolock();
		m_regPort.erase(pPortObj);
		m_lock.unlock();
	}

	bool IsAttached(IIocpPortBase* pPortObj) 
	{
		bool found = false;
		m_lock.dolock();
		found = m_regPort.end() != m_regPort.find(pPortObj);
		m_lock.unlock();
		return found;
	}

	BOOL BindHandle(HANDLE hHandle, IIocpPortBase* pPortObj)
	{
		if (!IsAttached(pPortObj))
			AttachPort(pPortObj);
		HANDLE hNewIocp = CreateIoCompletionPort(
			hHandle, m_hIocp, (ULONG_PTR)pPortObj, 0);
		if (hNewIocp == m_hIocp)
			return TRUE;
		PROMPT(_T("please create the IOCP. "));
		return FALSE;
	}

protected:
	inline int AdvisedThreadCount()
	{
		SYSTEM_INFO si;
		ZeroMemory(&si, sizeof(si));
		GetSystemInfo(&si);
		TRACE(_T("%d processor(s), advise %d thread(s)\n"),
			si.dwNumberOfProcessors, si.dwNumberOfProcessors*2+2);
		return (int)si.dwNumberOfProcessors * 2 + 2;
	}

protected:
	//worker thread
	class CWorkerThread : public threadbase
	{
	private:
		CWorkerThread(const CWorkerThread& other);
		CWorkerThread& operator=(const CWorkerThread& other);

	private:
		CIocpCore& r_iocp;

	public:
		CWorkerThread(CIocpCore& iocp)
			: r_iocp(iocp)
		{}

	public:
		virtual void unblock()
		{
			TRACE(_T("post quit status when destroy thread.\n"));
			PostQueuedCompletionStatus(r_iocp.m_hIocp, 0, 0, 0);
			__super::unblock();
		}

	protected:
		virtual bool pump()
		{
			TRACE(_T("thread id:0x%x pump\n"), threadid());
			// returns false may means a status with an error IO operation. 
			// so you must call OnIoCompletion with the error code if the status is normal. 
			// the err code of IO operation can be got from system last error. 
			DWORD dwBytesRet = 0;
			IIocpPortBase* pPortObj = 0;
			LPOVERLAPPED lpOvlp = NULL;
			BOOL ret = GetQueuedCompletionStatus(
				r_iocp.m_hIocp, &dwBytesRet, (ULONG_PTR*)&pPortObj, &lpOvlp, IOCP_TIMEOUT);
			DWORD errCode = 0;
			if (ret && !lpOvlp && !pPortObj) {
				ret = FALSE;
				SetLastError(ERROR_OPERATION_ABORTED);
			}
			if (!ret) {
				errCode = GetLastError(); //last error is valid only when ret is false
				TRACE(_T("thread id:0x%x got error status, err:%d\n"), threadid(), errCode);
				if (ERROR_OPERATION_ABORTED == errCode) {
					TRACE(_T("post quit status when destroy thread.\n"));
					PostQueuedCompletionStatus(r_iocp.m_hIocp, 0, 0, 0);
					return false;
				}
				if (WAIT_TIMEOUT == errCode) {
					TRACE(_T("thread id:0x%x on idle\n"), threadid());
					r_iocp.m_lock.dolock();
					std::set<IIocpPortBase*>::iterator it;
					for (it=r_iocp.m_regPort.begin(); it!=r_iocp.m_regPort.end(); ++it) {
						IIocpPortBase* pPortObj = *it;
						if (pPortObj)
							pPortObj->OnIocpTimeout(IOCP_TIMEOUT);
					}
					r_iocp.m_lock.unlock();
					return true;
				}
			}
			if (lpOvlp && pPortObj && r_iocp.IsAttached(pPortObj)) {
				TRACE(_T("thread id:0x%x call OnIoCompletion(with err %d)\n"), threadid(), errCode);
				pPortObj->OnIoCompletion(dwBytesRet, lpOvlp, errCode);
			}
			return true;
		}
	};

private:
	class CDaemonThread : public threadbase
	{
	private:
		CDaemonThread(const CDaemonThread& other);
		CDaemonThread& operator=(const CDaemonThread& other);

	private:
		CIocpCore& r_iocp;

	public:
		CDaemonThread(CIocpCore& iocp)
			: r_iocp(iocp)
		{}

	protected:
		//keep pause, scan thread pool by every pulse of idle. 
		virtual bool pump(){pause();return true;}

		//do idle to renew the dead threads. 
		virtual void idle()
		{
			TRACE(_T("thread monitor do idle\n"));
			r_iocp.m_lock.dolock();
			std::vector<HANDLE> handles;
			handles.resize((int)r_iocp.m_vecThread.size());
			for (int i=0; i<(int)r_iocp.m_vecThread.size(); ++i) {
				CWorkerThread* worker = r_iocp.m_vecThread[i];
				handles[i] = worker->safehandle();
			}
			//wait for any worker thread exit, no wait if all in normal. 
			int ret = WaitForMultipleObjects(handles.size(), &handles[0], FALSE, 0);
			int index = ret - WAIT_OBJECT_0;
			if (index >= 0 && index < (int)r_iocp.m_vecThread.size()) {
				TRACE(_T("worker thread broken on index:%d\n"), index);
				CWorkerThread* worker = r_iocp.m_vecThread[index];
				if (!worker)
					worker = new CWorkerThread(r_iocp);
				else 
					worker->destroy();
				worker->init(true, IOCP_TIMEOUT);
				TRACE(_T("worker thread reset on index:%d\n"), index);
			}
			r_iocp.m_lock.unlock();
		}
	};

protected:
	CDaemonThread* m_daemon;
	std::vector<CWorkerThread*> m_vecThread;
	std::set<IIocpPortBase*> m_regPort;
	critical_section m_lock; //lock for both the thread pool and the reg pool
	HANDLE m_hIocp;

};

}//namespace

//////////////////////////////////////////////////////////////////////////
// doc: GetQueuedCompletionStatus():
//
//   If the function dequeues a completion packet for a successful I/O operation
// from the completion port, the return value is nonzero. The function stores 
// information in the variables pointed to by the 
// lpNumberOfBytes, lpCompletionKey, and lpOverlapped parameters.
//   If *lpOverlapped is NULL and the function does not dequeue a completion packet 
// from the completion port, the return value is zero. The function does not store 
// information in the variables pointed to by the 
// lpNumberOfBytes and lpCompletionKey parameters. 
// To get extended error information, call GetLastError. If the function 
// did not dequeue a completion packet because the wait timed out, GetLastError 
// returns WAIT_TIMEOUT.
//   If *lpOverlapped is not NULL and the function dequeues a completion packet 
// for a failed I/O operation from the completion port, the return value is zero. 
// The function stores information in the variables pointed to by 
// lpNumberOfBytes, lpCompletionKey, and lpOverlapped. 
// To get extended error information, call GetLastError.
//////////////////////////////////////////////////////////////////////////

#endif //__CR_IOCP_WITH_THREAD_POOL__